package spring.cloud.tasks.tasks_assign_manager.task_item_assign;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import spring.cloud.tasks.common.BatchTaskInstanceInfo;
import spring.cloud.tasks.common.BatchTaskReStartInfo;
import spring.cloud.tasks.common.task_item_allot.OnlineExecutor;
import spring.cloud.tasks.common.task_item_allot.TaskItem;
import spring.cloud.tasks.tasks_assign_manager.schedule.BatchTaskInstanceManager;
import spring.cloud.tasks.tasks_assign_manager.schedule.CronExpressionContextManager;
import spring.cloud.tasks.tasks_assign_manager.schedule.TaskSignature;
import spring.cloud.tasks.tasks_assign_manager.schedule.TaskSignatureManager;
import spring.cloud.tasks.tasks_assign_manager.service.TaskItemAssignService;
import spring.cloud.tasks.tasks_assign_manager.task_item_assign.pick_up.executor.PickUpWhenExecutorLostTraffic;
import spring.cloud.tasks.tasks_assign_manager.task_item_assign.pick_up.executor.PickUpWhenExecutorOffline;
import spring.cloud.tasks.tasks_assign_manager.task_item_assign.pick_up.executor.PickUpWhenExecutorRecoverTraffic;
import spring.cloud.tasks.tasks_assign_manager.task_item_assign.pick_up.task.PickUpWhenTaskDisable;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 针对每个任务实例+任务分片项都需要提前建立分片项节点。具体执行器通过分片项节点进行续期以及状态以及和管理系统交互数据等等。
 */
@Slf4j
@Setter
@Getter
public class TaskItemAssign {

    protected CronExpressionContextManager cronExpressionContextManager;

    protected TaskItemAssignService taskItemAssignService;

    public TaskItemAssign(CronExpressionContextManager cronExpressionContextManager, TaskItemAssignService taskItemAssignService) {

        this.cronExpressionContextManager = cronExpressionContextManager;

        this.taskItemAssignService = taskItemAssignService;
    }

    /**
     * <pre>
     * 1根据enableTaskIdList计算出最近需要进行执行的任务。如果任务分配了但还没有到执行的时间 那么可以进行重新分配。
     * 2历史enableTaskIdList任务只要没有到达最大的timeoutDateTime那么则允许在任务分配列表上面，虽然该任务不进行分配，但是有可能任务还在执行，这些任务即使执行失败也不修改执行计划。
     * 3节点上线只能等任务下次分配。
     * 4节点下线可强制进行一次任务分配。这样做需要任务具备一个属性:下线的时候任务可以立即被重新分配（一天只执行一次 或者 有限的次数 那么允许重新分配，否则任务会被丢失执行）。
     * </pre>
     */
    public void taskItemAssign() throws Exception {
        String nameSpace = taskItemAssignService.getNameSpace();
        CuratorFramework curatorFramework = taskItemAssignService.getCuratorFramework();
        TaskSignatureManager taskSignatureManager = new TaskSignatureManager(nameSpace, curatorFramework);
        BatchTaskInstanceManager batchTaskInstanceManager = new BatchTaskInstanceManager(nameSpace, curatorFramework);
        batchTaskInstanceManager.init(curatorFramework);
        log.info("taskItemAssign start");

        try {

            List<OnlineExecutor> oldOnlineExecutorList = taskItemAssignService.getOldOnlineExecutorTaskItemAssignList();
            //$oldOnlineExecutorList:=oldOnlineExecutorList(use copy)的image对象
            List<OnlineExecutor> onlineExecutorList = TaskItemAssignUtils.copyExecutorList(oldOnlineExecutorList);
            //
            List<String> lastTaskIdList = TaskItemAssignUtils.getTaskIdList(curatorFramework);
            List<String> lastEnableTaskIdList = TaskItemAssignUtils.getEnableTaskIdList(curatorFramework, lastTaskIdList);
            taskSignatureManager.init(lastEnableTaskIdList);
            //
            List<OnlineExecutor> lastOnlineExecutorList = TaskItemAssignUtils.getLastOnlineExecutorList(curatorFramework);
            List<String> lastOnlineExecutorIdList = TaskItemAssignUtils.getOnlineExecutorIdList(lastOnlineExecutorList);
            List<OnlineExecutor> lastOnlineOwnTrafficExecutorList = TaskItemAssignUtils.getOwnTrafficOnlineExecutorList(lastOnlineExecutorList);
            List<String> lastOnlineOwnTrafficExecutorIdList = TaskItemAssignUtils.getOwnTrafficOnlineExecutorIdList(lastOnlineExecutorList);
            List<OnlineExecutor> lastOnlineDoNotOwnTrafficExecutorList = TaskItemAssignUtils.getDoNotOwnTrafficOnlineExecutorList(lastOnlineExecutorList);
            List<String> lastOnlineDoNotOwnTrafficExecutorIdList = TaskItemAssignUtils.getDoNotOwnTrafficOnlineExecutorIdList(lastOnlineExecutorList);
            //
            TaskItemAssignPresentSituation taskItemAssignPresentSituation = new TaskItemAssignPresentSituation();
            taskItemAssignPresentSituation.oldOnlineExecutorList = onlineExecutorList;//old对象
            //taskItemAssignPresentSituation.newOwnTrafficOnlineExecutorList = newOwnTrafficOnlineExecutorList;
            //taskItemAssignPresentSituation.newDoNotOwnTrafficOnlineExecutorList = newDoNotOwnTrafficOnlineExecutorList;
            //
            taskItemAssignPresentSituation.lastTaskIdList = lastTaskIdList;
            taskItemAssignPresentSituation.lastEnableTaskIdList = lastEnableTaskIdList;
            //
            taskItemAssignPresentSituation.lastOnlineExecutorIdList = lastOnlineExecutorIdList;
            taskItemAssignPresentSituation.lastOnlineOwnTrafficExecutorIdList = lastOnlineOwnTrafficExecutorIdList;
            taskItemAssignPresentSituation.lastOnlineDoNotOwnTrafficExecutorIdList = lastOnlineDoNotOwnTrafficExecutorIdList;
            //
            TaskItemAssignFixPoints taskItemAssignFixPoints = TaskItemAssignUtils.getTaskItemAssignFixPoints(taskItemAssignPresentSituation);
            List<String> taskIdList = taskItemAssignFixPoints.getTaskIdList();
            List<String> disableTaskIdList = taskItemAssignFixPoints.getDisableTaskIdList();//优先处理[1]
            List<String> enableTaskIdList = taskItemAssignFixPoints.getEnableTaskIdList();
            List<String> offlineExecutorIdList = taskItemAssignFixPoints.getOfflineExecutorIdList();
            List<String> onlineExecutorIdList = taskItemAssignFixPoints.getOnlineExecutorIdList();//按照新的算
            List<String> onlineDoNotOwnTrafficExecutorIdList = taskItemAssignFixPoints.getOnlineDoNotOwnTrafficExecutorIdList();//优先处理[2]
            List<String> onlineOwnTrafficExecutorIdList = taskItemAssignFixPoints.getOnlineOwnTrafficExecutorIdList();
            //更新执行器的支持的任务list
            TaskItemAssignUtils.updateIsOwnTrafficAndSupportedTaskIdList(onlineExecutorList, lastOnlineExecutorList);
            //在上个版本的任务分配基础上新增新上线的执行器
            TaskItemAssignUtils.addNewOnlineExecutorList(onlineExecutorList, lastOnlineExecutorList);
            //重新最新的在线以及在线拥有流量的执行器信息
            List<OnlineExecutor> ownTrafficOnlineExecutorList = TaskItemAssignUtils.getOwnTrafficOnlineExecutorList(onlineExecutorList);
            List<TaskItem> pickUpEdTaskItemList = new ArrayList<TaskItem>();
            List<TaskItem> discardTaskItemList = new ArrayList<TaskItem>();
            //nullPreferExecutorSettings 只是占位空对象
            PreferExecutorSettings nullPreferExecutorSettings = new PreferExecutorSettings();
            nullPreferExecutorSettings.taskIdToIsConfiguredPreferExecutorIdListMap = null;
            nullPreferExecutorSettings.taskIdToConfiguredPreferExecutorIdListMap = null;
            nullPreferExecutorSettings.taskIdToOwnTrafficOnlineExecutorListMap = null;
            for (String disableTaskId : disableTaskIdList) {
                PickUpWhenTaskDisable pickUp = new PickUpWhenTaskDisable(taskItemAssignService);
                pickUp.setTaskId(disableTaskId);
                pickUp.pickUp(curatorFramework,//
                        taskIdList,//
                        enableTaskIdList,//
                        pickUpEdTaskItemList,//
                        discardTaskItemList,//
                        onlineExecutorList,//
                        ownTrafficOnlineExecutorList,//
                        nullPreferExecutorSettings);
            }
            for (String offlineExecutorId : offlineExecutorIdList) {
                PickUpWhenExecutorOffline pickUp = new PickUpWhenExecutorOffline(taskItemAssignService);
                pickUp.setExecutorId(offlineExecutorId);
                pickUp.pickUp(curatorFramework,//
                        taskIdList,//
                        enableTaskIdList,//
                        pickUpEdTaskItemList,//
                        discardTaskItemList,//
                        onlineExecutorList,//
                        ownTrafficOnlineExecutorList,//
                        nullPreferExecutorSettings);
            }
            for (String onlineDoNotOwnTrafficExecutorId : onlineDoNotOwnTrafficExecutorIdList) {
                PickUpWhenExecutorLostTraffic pickUp = new PickUpWhenExecutorLostTraffic(taskItemAssignService);
                pickUp.setExecutorId(onlineDoNotOwnTrafficExecutorId);
                pickUp.pickUp(curatorFramework,//
                        taskIdList,//
                        enableTaskIdList,//
                        pickUpEdTaskItemList,//
                        discardTaskItemList,//
                        onlineExecutorList,//
                        ownTrafficOnlineExecutorList,//
                        nullPreferExecutorSettings);
            }
            //摘取Executor不支持的任务(前面进行各种摘取后才这个)
            for (OnlineExecutor onlineExecutor : onlineExecutorList) {
                List<TaskItem> taskItemList = onlineExecutor.getAssignedTaskItemList();
                Iterator<TaskItem> iterator = taskItemList.iterator();
                while (iterator.hasNext()) {
                    TaskItem taskItem = iterator.next();
                    if (!onlineExecutor.getSupportedTaskIdList().contains(taskItem.getTaskId())) {
                        onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel() - taskItem.getLoadLevel());
                        iterator.remove();
                        pickUpEdTaskItemList.add(taskItem);
                    }
                }
            }
            long nowDateTime = System.currentTimeMillis();
            {
                //清理超时的
                {//非批次任务处理
                    Iterator<TaskItem> iterator = pickUpEdTaskItemList.listIterator();
                    while (iterator.hasNext()) {
                        TaskItem taskItem = iterator.next();
                        String taskId = taskItem.getTaskId();
                        String taskType = taskItem.getTaskType();
                        String taskInstanceId = taskItem.getTaskInstanceId();
                        String taskRoundId = taskItem.getTaskRoundId();
                        if ((TaskItem.TASK_TYPE_ITEM_TASK.equals(taskType) || TaskItem.TASK_TYPE_MESSAGE_TASK.equals(taskType))) {
                            if (taskItem.getTimeoutDateTime().getTime() <= nowDateTime) {
                                iterator.remove();
                                continue;
                            }
                        } else if (TaskItem.TASK_TYPE_BATCH_TASK.equals(taskType)) {
                            //没有维护的批次任务超时清理
                            if (taskItem.getTimeoutDateTime().getTime() <= nowDateTime) {
                                iterator.remove();
                                continue;
                            }
                        } else {
                            throw new IllegalArgumentException();
                        }
                    }
                }
                {
                    for (OnlineExecutor onlineExecutor : onlineExecutorList) {
                        List<TaskItem> taskItemList = onlineExecutor.getAssignedTaskItemList();
                        Iterator<TaskItem> iterator = taskItemList.iterator();
                        while (iterator.hasNext()) {
                            TaskItem taskItem = iterator.next();
                            String taskId = taskItem.getTaskId();
                            String taskType = taskItem.getTaskType();
                            String taskInstanceId = taskItem.getTaskInstanceId();
                            String taskRoundId = taskItem.getTaskRoundId();
                            if ((TaskItem.TASK_TYPE_ITEM_TASK.equals(taskType) || TaskItem.TASK_TYPE_MESSAGE_TASK.equals(taskType))) {
                                if (taskItem.getTimeoutDateTime().getTime() <= nowDateTime) {
                                    onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel() - taskItem.getLoadLevel());
                                    iterator.remove();
                                    continue;
                                }
                            } else if (TaskItem.TASK_TYPE_BATCH_TASK.equals(taskType)) {
                                if (taskItem.getTimeoutDateTime().getTime() <= nowDateTime) {
                                    onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel() - taskItem.getLoadLevel());
                                    iterator.remove();
                                    continue;
                                }
                            } else {
                                throw new IllegalArgumentException();
                            }
                        }
                    }
                }
                {//批次任务处理
                    //如果批次任务实例信息不维护了 那么只需要根据超时时间进行处理即可
                    //找到所有的批次任务实例信息,进行超时处理 如果超时了那么后面则不进行状态更新，更改状态且移掉实例分片。
                }
            }

            //清理签名
            //任务启用,到了时间点就进行分配，根据waitedAssignTaskItemList计算出最近可以进行资源分配的任务列表
            List<TaskItem> waitedAssignTaskItemList = TaskItemAssignUtils.getWaitedAssignTaskItemList(curatorFramework, cronExpressionContextManager, lastEnableTaskIdList);
            Set<String> waitedAssignTaskIdSet = new HashSet<>();
            Map<String, String> waitedAssignTaskIdToInstanceIdMap = new HashMap<>();
            Map<String, BatchTaskInstanceInfo> waitedAssignTaskIdToBatchTaskInstanceInfoMap = new HashMap<>();
            for (TaskItem taskItem : waitedAssignTaskItemList) {
                waitedAssignTaskIdSet.add(taskItem.getTaskId());
                TaskSignature taskSignature = taskSignatureManager.taskIdToSignatureMap.get(taskItem.getTaskId());
                String signature = taskSignature.getSignature();
                taskItem.setSignature(signature);
                //
                if (taskItem.getTaskType().equals(TaskItem.TASK_TYPE_BATCH_TASK)) {
                    if (waitedAssignTaskIdToInstanceIdMap.containsKey(taskItem.getTaskId())) {
                        throw new IllegalStateException("同一个时间点只能新启动一个批次任务的实例");
                    }
                    waitedAssignTaskIdToInstanceIdMap.put(taskItem.getTaskId(), taskItem.getTaskInstanceId());//同一个时间点只能新启动一个批次任务的实例
                }
            }
            {//只要任务签名不进行改变，enable的任务只要不进行下次分配，那么一直可以被进行执行。
                Iterator<TaskItem> iterator = pickUpEdTaskItemList.listIterator();
                while (iterator.hasNext()) {
                    TaskItem taskItem = iterator.next();
                    String taskId = taskItem.getTaskId();
                    String taskType = taskItem.getTaskType();
                    String taskInstanceId = taskItem.getTaskInstanceId();
                    String taskRoundId = taskItem.getTaskRoundId();
                    if (waitedAssignTaskIdSet.contains(taskId) && (TaskItem.TASK_TYPE_ITEM_TASK.equals(taskType) || TaskItem.TASK_TYPE_MESSAGE_TASK.equals(taskType))) {
                        TaskSignature taskSignature = taskSignatureManager.taskIdToSignatureMap.get(taskId);
                        String signature = taskSignature.getSignature();
                        if (!signature.equals(taskItem.getSignature())) {
                            iterator.remove();
                            continue;
                        }
                    } else if (TaskItem.TASK_TYPE_BATCH_TASK.equals(taskType)) {
                        boolean isMaintain = batchTaskInstanceManager.isMaintain(taskId, taskInstanceId, taskRoundId);//批次任务超时的代码在后面
                        if (!isMaintain) {
                            iterator.remove();
                            continue;
                        }
                    }
                }
            }
            //
            {
                for (OnlineExecutor onlineExecutor : onlineExecutorList) {
                    List<TaskItem> assignedTaskItemList = onlineExecutor.getAssignedTaskItemList();
                    Iterator<TaskItem> iterator = assignedTaskItemList.listIterator();
                    while (iterator.hasNext()) {
                        TaskItem taskItem = iterator.next();
                        String taskId = taskItem.getTaskId();
                        String taskType = taskItem.getTaskType();
                        String taskInstanceId = taskItem.getTaskInstanceId();
                        String taskRoundId = taskItem.getTaskRoundId();
                        if (waitedAssignTaskIdSet.contains(taskId) && (TaskItem.TASK_TYPE_ITEM_TASK.equals(taskType) || TaskItem.TASK_TYPE_MESSAGE_TASK.equals(taskType))) {
                            TaskSignature taskSignature = taskSignatureManager.taskIdToSignatureMap.get(taskId);
                            String signature = taskSignature.getSignature();
                            if (!signature.equals(taskItem.getSignature())) {
                                onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel() - taskItem.getLoadLevel());
                                iterator.remove();
                                continue;
                            }
                        } else if (TaskItem.TASK_TYPE_BATCH_TASK.equals(taskType)) {
                            boolean isMaintain = batchTaskInstanceManager.isMaintain(taskId, taskInstanceId, taskRoundId);//批次任务超时的代码在后面
                            if (!isMaintain) {
                                onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel() - taskItem.getLoadLevel());
                                iterator.remove();
                                continue;
                            }
                        }
                    }
                }
            }
            ConcurrentHashMap<String, List<BatchTaskReStartInfo>> taskIdToBatchTaskReStartInfoListMap = batchTaskInstanceManager.taskIdToBatchTaskReStartInfoListMap;
            ConcurrentHashMap<String, List<BatchTaskInstanceInfo>> taskIdToBatchTaskInstanceInfoListMap = batchTaskInstanceManager.taskIdToBatchTaskInstanceInfoListMap;
            for (Map.Entry<String, List<BatchTaskReStartInfo>> entry : taskIdToBatchTaskReStartInfoListMap.entrySet()) {
                List<BatchTaskReStartInfo> batchTaskReStartInfoList = entry.getValue();
                for (BatchTaskReStartInfo batchTaskReStartInfo : batchTaskReStartInfoList) {
                    BatchTaskInstanceInfo batchTaskInstanceInfo = TaskItemAssignUtils.rebuildBatchTaskTaskItem(batchTaskReStartInfo);
                    if (batchTaskInstanceInfo.isRestartAssignTaskItemPlanCreate()) {
                        //需要先将现在的执行计划终止掉
                        {
                            Iterator<TaskItem> iterator = pickUpEdTaskItemList.listIterator();
                            while (iterator.hasNext()) {
                                TaskItem taskItem = iterator.next();
                                String taskId = taskItem.getTaskId();
                                String taskType = taskItem.getTaskType();
                                String taskInstanceId = taskItem.getTaskInstanceId();
                                String taskRoundId = taskItem.getTaskRoundId();
                                if (TaskItem.TASK_TYPE_BATCH_TASK.equals(taskType) && taskId.equals(batchTaskInstanceInfo.getTaskId()) &&//
                                        (taskInstanceId.equals(batchTaskInstanceInfo.getTaskInstanceId())) &&//
                                        (taskRoundId.equals(batchTaskInstanceInfo.getTaskRoundId()))//
                                ) {
                                    iterator.remove();
                                    continue;
                                }
                            }
                        }
                        for (OnlineExecutor onlineExecutor : onlineExecutorList) {
                            List<TaskItem> assignedTaskItemList = onlineExecutor.getAssignedTaskItemList();
                            Iterator<TaskItem> iterator = assignedTaskItemList.listIterator();
                            while (iterator.hasNext()) {
                                TaskItem taskItem = iterator.next();
                                String taskId = taskItem.getTaskId();
                                String taskType = taskItem.getTaskType();
                                String taskInstanceId = taskItem.getTaskInstanceId();
                                String taskRoundId = taskItem.getTaskRoundId();
                                if (TaskItem.TASK_TYPE_BATCH_TASK.equals(taskType) &&//
                                        taskId.equals(batchTaskInstanceInfo.getTaskId()) &&//
                                        (taskInstanceId.equals(batchTaskInstanceInfo.getTaskInstanceId())) &&//
                                        (taskRoundId.equals(batchTaskInstanceInfo.getTaskRoundId()))//
                                ) {
                                    onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel() - taskItem.getLoadLevel());
                                    iterator.remove();
                                    continue;
                                }
                            }
                        }
                        //增加新的
                        pickUpEdTaskItemList.addAll(batchTaskInstanceInfo.getTaskItemList());
                    }
                }
            }
            for (Map.Entry<String, List<BatchTaskInstanceInfo>> entry : taskIdToBatchTaskInstanceInfoListMap.entrySet()) {
                List<BatchTaskInstanceInfo> batchTaskInstanceInfoList = entry.getValue();
                for (BatchTaskInstanceInfo batchTaskInstanceInfo : batchTaskInstanceInfoList) {
                    TaskItemAssignUtils.rebuildBatchTaskTaskItem(batchTaskInstanceInfo);
                    Boolean rebuildAssignTaskItemPlanCreate = batchTaskInstanceInfo.getRebuildAssignTaskItemPlanCreate();
                    if (rebuildAssignTaskItemPlanCreate == null) {
                        continue;//什么也不做
                    } else {
                        //rebuildAssignTaskItemPlanCreate!=null 无论哪种情况true/false都需要先将现在的执行计划终止掉
                        //true删掉老的 增加新的
                        //false删掉老的 任务结束
                        {
                            Iterator<TaskItem> iterator = pickUpEdTaskItemList.listIterator();
                            while (iterator.hasNext()) {
                                TaskItem taskItem = iterator.next();
                                String taskId = taskItem.getTaskId();
                                String taskType = taskItem.getTaskType();
                                String taskInstanceId = taskItem.getTaskInstanceId();
                                String taskRoundId = taskItem.getTaskRoundId();
                                if (TaskItem.TASK_TYPE_BATCH_TASK.equals(taskType) &&//
                                        taskId.equals(batchTaskInstanceInfo.getTaskId()) &&//
                                        (taskInstanceId.equals(batchTaskInstanceInfo.getTaskInstanceId())) &&//
                                        (taskRoundId.equals(batchTaskInstanceInfo.getTaskRoundId()))//
                                ) {
                                    iterator.remove();
                                    continue;
                                }
                            }
                        }
                        for (OnlineExecutor onlineExecutor : onlineExecutorList) {
                            List<TaskItem> assignedTaskItemList = onlineExecutor.getAssignedTaskItemList();
                            Iterator<TaskItem> iterator = assignedTaskItemList.listIterator();
                            while (iterator.hasNext()) {
                                TaskItem taskItem = iterator.next();
                                String taskId = taskItem.getTaskId();
                                String taskType = taskItem.getTaskType();
                                String taskInstanceId = taskItem.getTaskInstanceId();
                                String taskRoundId = taskItem.getTaskRoundId();
                                if (TaskItem.TASK_TYPE_BATCH_TASK.equals(taskType) &&
                                        taskId.equals(batchTaskInstanceInfo.getTaskId()) &&
                                        (taskInstanceId.equals(batchTaskInstanceInfo.getTaskInstanceId())) &&
                                        (taskRoundId.equals(batchTaskInstanceInfo.getTaskRoundId()))
                                ) {
                                    onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel() - taskItem.getLoadLevel());
                                    iterator.remove();
                                    continue;
                                }
                            }
                        }
                        //为true代表更新下一阶段计划
                        if (rebuildAssignTaskItemPlanCreate) {
                            //清理老的执行计划 创建新的
                            pickUpEdTaskItemList.addAll(batchTaskInstanceInfo.getTaskItemList());
                        } else {
                            //清理老的执行计划 任务结束(代码有点重复)
                        }
                    }

                }
            }
            //新的批次信息增加。
            //针对分片任务和MQ任务，如果分片信息存在那么把自己删除，否则添加到pickUpEdTaskItemList。针对批次任务则可以直接添加。
            Iterator<TaskItem> waitedAssignTaskItemListIterator = waitedAssignTaskItemList.iterator();
            while (waitedAssignTaskItemListIterator.hasNext()) {
                TaskItem taskItem = waitedAssignTaskItemListIterator.next();
                if (taskItem.getTaskType().equals(TaskItem.TASK_TYPE_ITEM_TASK) || taskItem.getTaskType().equals(TaskItem.TASK_TYPE_MESSAGE_TASK)) {
                    boolean include = false;
                    for (TaskItem taskItem1 : pickUpEdTaskItemList) {
                        if (taskItem1.getTaskId().equals(taskItem.getTaskId()) && //
                                taskItem1.getTaskInstanceId().equals(taskItem.getTaskInstanceId()) && //
                                taskItem1.getTaskRoundId().equals(taskItem.getTaskRoundId()) //
                        ) {
                            include = true;
                            break;
                        }
                    }
                    if (include) {
                        waitedAssignTaskItemListIterator.remove();
                        continue;
                    }
                    for (OnlineExecutor onlineExecutor : onlineExecutorList) {
                        for (TaskItem taskItem1 : onlineExecutor.getAssignedTaskItemList()) {
                            if (taskItem1.getTaskId().equals(taskItem.getTaskId()) &&//
                                    taskItem1.getTaskInstanceId().equals(taskItem.getTaskInstanceId()) &&//
                                    taskItem1.getTaskRoundId().equals(taskItem.getTaskRoundId())//
                            ) {
                                include = true;
                                break;
                            }
                        }
                    }
                    if (include) {
                        waitedAssignTaskItemListIterator.remove();
                        continue;
                    }
                    pickUpEdTaskItemList.add(taskItem);
                    waitedAssignTaskItemListIterator.remove();
                    continue;
                } else if (taskItem.getTaskType().equals(TaskItem.TASK_TYPE_BATCH_TASK)) {
                    pickUpEdTaskItemList.add(taskItem);//批次任务直接添加，因为都是新的实例
                    waitedAssignTaskItemListIterator.remove();
                    continue;
                } else {
                    throw new IllegalArgumentException();
                }

            }//waitedAssignTaskItemList is empty

            //剩下的是有效任务 以及在线执行器(可能有的执行器没有分配流量通过)
            PreferExecutorSettings preferExecutorSettings = new PreferExecutorSettings();
            for (String taskId : enableTaskIdList) {
                preferExecutorSettings.taskIdToIsConfiguredPreferExecutorIdListMap.put(taskId, TaskItemAssignUtils.preferExecutorIdListIsConfigured(curatorFramework, taskId));
                preferExecutorSettings.taskIdToConfiguredPreferExecutorIdListMap.put(taskId, TaskItemAssignUtils.getPreferExecutorIdListConfigured(curatorFramework, taskId, onlineExecutorIdList));
                preferExecutorSettings.taskIdToOwnTrafficOnlineExecutorListMap.put(taskId, TaskItemAssignUtils.getSupportedTaskIdOwnTrafficOnlineExecutorList(ownTrafficOnlineExecutorList, taskId));
            }
            //将配置没有优先执行器list的任务或者不在优先执行器list执行器节点清除(非优选节点可以不用忙进行切换)
            TaskItemAssignUtils.fixPreferOnlineExecutorList(enableTaskIdList, preferExecutorSettings, pickUpEdTaskItemList, ownTrafficOnlineExecutorList);
            //检查哪些执行器拥有流量但是没有进行任务分配,然后进行执行器恢复流量
            List<OnlineExecutor> onlineOwnTrafficButNotAssignedExecutorList = TaskItemAssignUtils.getOnlineOwnTrafficButNotAssignedExecutorList(onlineExecutorList);
            for (OnlineExecutor onlineExecutor : onlineOwnTrafficButNotAssignedExecutorList) {
                PickUpWhenExecutorRecoverTraffic pickUp = new PickUpWhenExecutorRecoverTraffic(taskItemAssignService);
                pickUp.setExecutorId(onlineExecutor.getExecutorId());
                pickUp.pickUp(curatorFramework,//
                        taskIdList,//
                        enableTaskIdList,//
                        pickUpEdTaskItemList,//
                        discardTaskItemList,//
                        onlineExecutorList,//
                        ownTrafficOnlineExecutorList,//
                        preferExecutorSettings);
            }
            //等流量分配均匀后再进行下次任务调度分配  这里的逻辑可能是错误的 因为有的分配不需要均衡，可能还需要丢弃摘取一次。比如批次任务，状态阶段变更。
            boolean pickUpEd = true;
            if (pickUpEd) {
                // 放回
                putBack(curatorFramework, taskIdList, enableTaskIdList, pickUpEdTaskItemList, onlineExecutorList, ownTrafficOnlineExecutorList,//
                        preferExecutorSettings
                );
                // 持久化
                //创建对于的任务item列表
                //删除老的 新增新的 任务/实例/轮次/taskItemIndex list
                taskItemAssignService.fixTaskItemIndexNode(onlineExecutorList);
                taskItemAssignService.persistOnlineExecutorTaskItemAssignList(onlineExecutorList);
                for (Map.Entry<String, String> entry : waitedAssignTaskIdToInstanceIdMap.entrySet()) {
                    String taskId = entry.getKey();
                    String instanceId = entry.getValue();
                    BatchTaskInstanceInfo batchTaskInstanceInfo = waitedAssignTaskIdToBatchTaskInstanceInfoMap.get(taskId);//一次分配一个任务只会有一个批次产生
                    batchTaskInstanceManager.taskIdToBatchTaskInstanceInfoListMap.putIfAbsent(taskId, new ArrayList<>());
                    List<BatchTaskInstanceInfo> batchTaskInstanceInfoList = batchTaskInstanceManager.taskIdToBatchTaskInstanceInfoListMap.get(taskId);
                    batchTaskInstanceInfoList.add(batchTaskInstanceInfo);
                }
            }
        } catch (InterruptedException e) {
            log.info("{}-{} {} is interrupted", taskItemAssignService.getNameSpace(), taskItemAssignService.getHost(), this.getClass().getSimpleName());
            Thread.currentThread().interrupt();
        } catch (Throwable t) {
            log.error(t.getMessage(), t);
            // 如果需要全量分片，且当前线程不是全量分片线程，则直接返回，没必要做分片
            boolean isNameSpaceTasksTaskItemAssign = false;// this instanceof NameSpaceTasksTaskItemAssign;
            if (taskItemAssignService.isNeedAllSharding() && !isNameSpaceTasksTaskItemAssign) {
                log.info("the {} will be ignored, because there will be {}", this.getClass().getSimpleName(), NameSpaceTasksTaskItemAssign.class.getSimpleName());
                return;
            }
            if (!isNameSpaceTasksTaskItemAssign) { // 如果当前不是全量分片，则需要全量分片来拯救异常
                taskItemAssignService.setNeedAllSharding(true);
                new NameSpaceTasksTaskItemAssign(taskItemAssignService);//.taskItemAssign();
            }
            if (isNameSpaceTasksTaskItemAssign) { // 如果是全量分片，不再进行全量分片
                taskItemAssignService.setNeedAllSharding(false);
            }
        } finally {
        }
        log.info("taskItemAssign end");
    }


    /**
     * 放回算法：拿取Shard，放进负荷最小的executor
     */
    protected void putBack(CuratorFramework curatorFramework, List<String> taskIdList, List<String> enableTaskIdList, List<TaskItem> pickUpEdTaskItemList, List<OnlineExecutor> onlineExecutorList, List<OnlineExecutor> ownTrafficOnlineExecutorList,//
                           PreferExecutorSettings preferExecutorSettings) throws Exception {
        if (onlineExecutorList.isEmpty()) {
            log.warn("Unnecessary to put shards back to executors balanced because of no executor");
            return;
        }
        TaskItemAssignUtils.sortTaskItemList(pickUpEdTaskItemList);
        // 1、放回配置了preferList的Shard
        TaskItemAssignUtils.putBackWithPreferExecutorIdListConfigured(pickUpEdTaskItemList, preferExecutorSettings.taskIdToOwnTrafficOnlineExecutorListMap, preferExecutorSettings.taskIdToIsConfiguredPreferExecutorIdListMap, preferExecutorSettings.taskIdToConfiguredPreferExecutorIdListMap);
        // 2、放回没有配置preferList的Shard
        TaskItemAssignUtils.putBackWithoutPreferExecutorIdListConfigured(pickUpEdTaskItemList, preferExecutorSettings.taskIdToOwnTrafficOnlineExecutorListMap);
    }


}
